package que.app.dwd.db;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import que.app.SqlApp;
import que.common.Word;
import que.util.SQLUtil;

/**
 * @author Naruto
 * @description: 一条数据为一人的一次考试的答卷信息
 * @Class_Name que.app.dwd.db.DwdExamInfo
 * @Date 2022/09/02 23:10
 */
public class DwdExamInfo extends SqlApp {
    public static void main(String[] args) {
        new DwdExamInfo().init(
                2,
                9006,
        "DwdExamInfo");
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"DwdExamInfo");

        //一张答卷
        Table testExam = tEnv.sqlQuery(
                "select " +
                        "data['id'] id, " +
                        "data['user_id'] user_id, " +
                        "data['paper_id'] paper_id, " +
                        "data['score'] score, " +
                        "data['duration_sec'] duration_sec, " +
                        "ts, " +
                        "pt " +
                        "from ods_db " +
                        "where `database` = 'edu' " +
                        "and `table` = 'order_info' " +
                        "and ( " +
                        "   `type` = 'insert'"
        );

        tEnv.createTemporaryView("test_exam",testExam);

        //获取试卷表信息
        tEnv.executeSql("create table test_papaer(" +
                "id bigint, " +
                "paper_title string, " +
                "course_id bigint " +
                ")with(" +
                " 'connector'='jdbc',  " +
                " 'url' = 'jdbc:mysql://hadoop162:3306/edu?useSSL=false'," +
                " 'table-name' = 'test_papaer', " +
                " 'username' = 'root',  " +
                " 'password' = 'aaaaaa'," +
                " 'lookup.cache.max-rows' = '10', " +
                " 'lookup.cache.ttl' = '1 hour' " +
                ")");

        //课程信息信息
        tEnv.executeSql("create table course_info(" +
                "id bigint, " +
                "course_name string, " +
                "course_id bigint " +
                ")with(" +
                " 'connector'='jdbc',  " +
                " 'url' = 'jdbc:mysql://hadoop162:3306/edu?useSSL=false'," +
                " 'table-name' = 'course_info', " +
                " 'username' = 'root',  " +
                " 'password' = 'aaaaaa'," +
                " 'lookup.cache.max-rows' = '10', " +
                " 'lookup.cache.ttl' = '1 hour' " +
                ")");

        //维度退化
        Table result = tEnv.sqlQuery(
                "select " +
                        "t.id, " +
                        "t.user_id, " +
                        "t.paper_id, " +
                        "p.paper_title, " +
                        "c.course_name, " +
                        "t.score, " +
                        "t.duration_sec, " +
                        "t.ts " +
                        "from test_exam t " +
                        "join test_papaer for system_time as of t.pt as p " +
                        "on t.paper_id=p.id " +
                        "join course_info for system_time as of t.pt as c " +
                        "on t.paper_id=c.id "
        );

        //定义一个动态表语kafka的topic关联
        tEnv.executeSql(
                "create table dwd_exam_info(" +
                        "t.id bigint, " +
                        "t.user_id bigint, " +
                        "t.paper_id bigint, " +
                        "p.paper_title string, " +
                        "c.course_name string, " +
                        "t.score bigint, " +
                        "t.duration_sec bigint, " +
                        "t.ts bigint" +
                        ")" + SQLUtil.getKafkaSink(Word.TOPIC_DWD_EXAM_INFO)
        );

        result.executeInsert("dwd_exam_info");
    }






}
